package com.taobao.analyse.main;


import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;

/**
 * 增量编辑记录扫描入口
 * @author xuanwei
 * @version 2012-3-15
 * @Mail-To xuanwei@taobao.com
 */
public class IncrMain {
	private String	baseOutPath	= "Your path";

	public JobConf getJob(String[] args) throws Exception {
		JobConf conf = new JobConf(new Configuration(), IncrMain.class);
		// MISC
		conf.setJobName("LCParser");
		conf.set("mapred.child.java.opts", "-Xmx2G");
		conf.setMaxMapAttempts(2);
		conf.setNumReduceTasks(1);

		// MAPPER
		conf.setMapOutputKeyClass(LongWritable.class);
		conf.setMapOutputValueClass(Text.class);

		// REDUCER
		conf.setOutputKeyClass(LongWritable.class);
		conf.setOutputValueClass(Text.class);
		//reducerδ�趨
		//conf.setReducerClass(JZLifeCycleReducer.class);

		// INPUTδ�趨
		//MultipleInputs.addInputPath(conf, new Path("/group/tbauction/lifecycle/incr/pt=20120314000000"),SequenceFileInputFormat.class, JZLifeCycleMapper.class);

		// OUTPUT
		Path out = this.buildOutputPath();
		conf.setOutputFormat(TextOutputFormat.class);

		System.out.println("�������·��:" + out.toString());
		SequenceFileOutputFormat.setOutputPath(conf, out);
		return conf;
	}

	private Path buildOutputPath() {
		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");
		String today = dateFormat.format(new Date());

		return new Path(baseOutPath + "pt=" + today + "000000/");
	}

	public static void main(String[] args) throws Exception {
		try {
			JobClient.runJob(new IncrMain().getJob(args));
		} catch (Exception e) {
			throw e;
		}
	}
}
